package com.atguigu.gmall.realtime.utils;

import com.atguigu.gmall.realtime.beans.TransientSink;
import com.atguigu.gmall.realtime.common.GmallConfig;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcExecutionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

import java.lang.reflect.Field;
import java.sql.PreparedStatement;
import java.sql.SQLException;

/**
 * Author: Felix
 * Date: 2021/12/6
 * Desc: 操作ClickHouse的工具类
 */
public class ClickHouseUtil {
    public static <T>SinkFunction<T> getSinkFunction(String sql){
        SinkFunction<T> sinkFunction = JdbcSink.<T>sink(
                sql,
                new JdbcStatementBuilder<T>() {
                    @Override
                    public void accept(PreparedStatement ps, T obj) throws SQLException {
                        //给sql中的问号占位符赋值  insert into visitor_stats_0609 values(?,?,?,?,?,?,?,?,?,?,?,?)
                        //获取流中对象的类型  --- 通过类型获取类中的属性
                        Field[] fieldArr = obj.getClass().getDeclaredFields();
                        //对属性进行遍历
                        int skipNum = 0;
                        for (int i = 0; i < fieldArr.length; i++) {
                            //每遍历一次  获取一个属性对象
                            Field field = fieldArr[i];

                            //判断当前属性是否需要保存到CK中
                            TransientSink transientSink = field.getAnnotation(TransientSink.class);
                            if(transientSink != null){
                                skipNum ++;
                                //如果对象不为空 ，说明属性上有@TransientSink注解标注，不需要保存到CK中
                                continue;
                            }

                            //设置属性的访问权限
                            field.setAccessible(true);
                            try {
                                //获取流中对象对应的属性值   正常:对象.属性    反射: 属性对象.get(对象)
                                Object fieldValue = field.get(obj);
                                //给问号占位符赋值
                                ps.setObject(i + 1 - skipNum , fieldValue );
                            } catch (IllegalAccessException e) {
                                e.printStackTrace();
                            }
                        }
                    }
                },
                JdbcExecutionOptions.builder()
                        .withBatchSize(5)
                        .build(),
                new JdbcConnectionOptions.JdbcConnectionOptionsBuilder()
                        .withDriverName("ru.yandex.clickhouse.ClickHouseDriver")
                        .withUrl(GmallConfig.CLICKHOUSE_URL)
                        .build()
        );
        return sinkFunction;
    }
}